
#include "BondMarketStreamV2.h"
#include "RealTimeDealUtil.h"
#include "RealTimeDealList.h"
#include <core/time/date.h>
#include <uam/ServerTimeMgr.h>
#include <uam/UserAccountManager.h>
#include "RealTimeDealStat.h"
#include <QtGlobal>

typedef MarketStreamRec			Stat;
typedef MarketStreamInfo		Record;

#ifndef BASSERTS
#define BASSERTS //
#endif // !BASSERTS

using namespace detail;
namespace detail{
	// compatible
	bool IsExistInVec(const std::vector<uint32_t>& vec, int ncount, uint32_t nvalue)
	{
		for (int i = 0; i < ncount && i <(int) vec.size(); i++)
		{
			if (vec[i] == nvalue)
				return true;
		}
		return false;
	}
}
	bool CMarketStream::IsBondPassType(const CBondInfo& bond, MarktType type)
	{
		if (bond.IsValidBond())
		{
			if (type == CMarketStream::MarktAll)
				return true;
			if (type == CMarketStream::MarktMunicp && bond.IsMunicipal())
				return true;
			else if (type == MSUtil::GetMarketStreamType(bond.GetBondSubType(), bond.GetRateType()))
			{
				if (type == CMarketStream::MarktFinance || 
					type == CMarketStream::MarktNFRate ||
					type == CMarketStream::MarktCountry || 
					type == CMarketStream::MarktShort ||
					type == CMarketStream::MarktChina ||
					type == CMarketStream::MarktEnterpr ||
					type == CMarketStream::MarktOther
					)
				return true;
			}
		}
		return false;
	}
	bool CMarketStream::IsBondPassPage(const CBondInfo& bond, emMktStreamPage pageType)
	{
		bool bValid = false;
		switch (pageType)
		{
		case CMarketStream::mktPage_Interest:
			if (bond.IsRatingDebt()) bValid = true;
			break;
		case CMarketStream::mktPage_Credit:
			if (!bond.IsRatingDebt() && !bond.IsBondNCD()) bValid = true;
			break;
		case CMarketStream::mktPage_TFRelative:
			if (MSUtil::IsBondRelativeType(bond) && MSUtil::IsBondRelativeTerm(bond)) bValid = true;
			break;
		case CMarketStream::mktPage_NCD:
			if (bond.IsBondNCD()) bValid = true;
			break;
		case  CMarketStream::mktPage_Credit_Inc_NCD:
			if (!bond.IsRatingDebt()) bValid = true;
			break;
		default:
			break;
		}
		return bValid;
	}
	int CMarketStream::GetMarketInfoKey(int stat, int mktType, const CBondInfo& bond)
	{
		int key = -1;
		if (stat == kStatTypeByType)
		{
			if (mktType == CMarketStream::MarktCountry || mktType == CMarketStream::MarktFinance || mktType == CMarketStream::MarktShort
				|| mktType == CMarketStream::MarktChina || mktType == CMarketStream::MarktEnterpr || mktType == CMarketStream::MarktOther){
				key = 1;
			}
		}
		else if (stat == CMarketStream::kStatTypeByDate)
		{
			std::string str = bond.GetRemainPeriod(ServerTimeMgr::instance().serverTime());
			int ldays = (int)MSUtil::CalcRemainingDay(str);

			if (ldays < 180)
				key = CMarketStream::DateRange1;

			else if (ldays >= 180 && ldays < 365)
				key = CMarketStream::DateRange2;

			else if (ldays >= 365 && ldays < 365 * 3)
				key = CMarketStream::DateRange3;

			else if (ldays >= 365 * 3 && ldays < 365 * 5)
				key = CMarketStream::DateRange4;

			else if (ldays >= 365 * 5 && ldays < 365 * 7)
				key = CMarketStream::DateRange5;

			else if (ldays >= 365 * 7 && ldays < 365 * 10)
				key = CMarketStream::DateRange6;

			else if (ldays >= 365 * 10)
				key = CMarketStream::DateRange7;
		}
		else if (stat == CMarketStream::kStatTypeTransDate)
		{
			std::string str = bond.GetRemainPeriod(ServerTimeMgr::instance().serverTime());
			int ldays = (int)MSUtil::CalcRemainingDay(str);
			if (mktType == CMarketStream::MarktShort){
				if (ldays < 30)
					key = CMarketStream::DateRange1;
				else if (ldays >= 30 && ldays < 90)
					key = CMarketStream::DateRange2;
				else if (ldays >= 90 && ldays < 180)
					key = CMarketStream::DateRange3;
				else if (ldays >= 180 && ldays < 270)
					key = CMarketStream::DateRange4;
				else if (ldays >= 270 && ldays < 365)
					key = CMarketStream::DateRange5;
			}
			else
			{
				if (ldays < 180)
					key = CMarketStream::DateRange1;
				else if (ldays >= 180 && ldays < 365)
					key = CMarketStream::DateRange2;
				else if (ldays >= 365 && ldays < 365 * 3)
					key = CMarketStream::DateRange3;
				else if (ldays >= 365 * 3 && ldays < 365 * 5)
					key = CMarketStream::DateRange4;
				else if (ldays >= 365 * 5 && ldays < 365 * 7)
					key = CMarketStream::DateRange5;
				else if (ldays >= 365 * 7 && ldays < 365 * 10)
					key = CMarketStream::DateRange6;
				else if (ldays >= 365 * 10)
					key = CMarketStream::DateRange7;
			}
		}
		else if (stat == CMarketStream::kStatTypeRatioType)
		{
			if (strcmp(bond.GetBondSubType(), "SCB") == 0){
				key = CMarketStream::MarktCenter;
			}
			else if (strcmp(bond.GetBondSubType(), "TLB") == 0 || strcmp(bond.GetBondSubType(), "LLB") == 0){
				key = CMarketStream::MarktLocal;
			}
			else if (mktType == CMarketStream::MarktCountry || mktType == CMarketStream::MarktFinance || mktType == CMarketStream::MarktShort
				|| mktType == CMarketStream::MarktChina || mktType == CMarketStream::MarktEnterpr || mktType == CMarketStream::MarktOther){
				key = mktType;
			}
		}
		else if (stat == kStatTypeTransRating)
		{
			const char* strBondRaing = bond.GetIssuerRating();
			if (strcmp(strBondRaing, "AAA") == 0)
				key = CMarketStream::RatingAAA;
			else if (strcmp(strBondRaing, "AA+") == 0)
				key = CMarketStream::RatingAAP;
			else if (strcmp(strBondRaing, "AA") == 0)
				key = CMarketStream::RatingAA;
			else if (strcmp(strBondRaing, "AA-") == 0)
				key = CMarketStream::RatingAAN;
			else if (strcmp(strBondRaing, "A+") == 0)
				key = CMarketStream::RatingAP;
			else if (strcmp(strBondRaing, "A") == 0)
				key = CMarketStream::RatingA;
			else key = CMarketStream::RatingOther;
		}
		else if (stat == CMarketStream::kStatTypeTransNFRate)
		{
			const char* strBondRaing = bond.GetRateType();
			if (strcmp(strBondRaing, "SHIBOR") == 0)	key = CMarketStream::RatypeShibor;
			else if (strcmp(strBondRaing, "DEPO") == 0)	key = CMarketStream::RatypeDepo;
		}
		return key;
	}
	bool CMarketStream::IsDealValid(const xQBMarketStreamUnit_c& msu, uint32_t& unDay, kStreamType& type)
	{
		// 用于导入成交数据时的过滤判断
		// 只有有效的符合要求的成交数据才会导入
		// 内部函数,不对外公开
		if (0 != strcmp("2", msu.m_type))
		{
			log_warning("[STREAM][INVALID]非成交信息 m_type:%s %s indexID:%d",msu.m_type,msu.m_company_id/*,msu.m_company_name*/,msu.m_indexID);
			return false;
		}
		unDay = qb::base::IntDate::timeToInt(qMax(msu.m_create_time, msu.m_modify_time));
		if (unDay <= 0)
		{
			log_warning("[STREAM][INVALID]非法创建时间或者更改时间 create:%64u  modify:%64u indexID:%d", msu.m_create_time, msu.m_modify_time,msu.m_indexID);
			return false;
		}
		if (strcmp(msu.m_company_id, "e") == 0)
		{
			type = CMarketStream::kStreamTypeExchange;
			return true;
		}
		if (strcmp(msu.m_company_id, "c") == 0)
		{
			type = type = CMarketStream::kStreamTypeCFets;
			return true;
		}
		int nBondIndex = CBondContainer::instance().GetBondIndex(msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket);
		if (!CBondContainer::instance().IsValidIndex(nBondIndex))
		{
			log_warning("[STREAM][INVALID]找不到债券 key:%s.%s indexID:%d", msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket, msu.m_indexID);
			return false;
		}
		if (CBondContainer::instance().ElementAtR(nBondIndex).IsBondNCD())
		{
			bool bNCDValid = SSUserAccountManager::instance().IsBrokerNCDAvailable(msu.m_company_id);
			if (!bNCDValid)
			{
				log_warning("[STREAM][INVALID]NCD不可用 key:%s.%s indexID:%d companyid:%s", msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket, msu.m_indexID,msu.m_company_id);
				return false;
			}
		}
		else
		{
			bool bBondValid = SSUserAccountManager::instance().IsBrokerBondAvailable(msu.m_company_id);
			if (!bBondValid)
			{
				log_warning("[STREAM][INVALID]Broker债券不允许 key:%s.%s indexID:%d companyid:%s", msu.m_body2.m_bondkey, msu.m_body2.m_listedmarket, msu.m_indexID, msu.m_company_id);
				return false;
			}
		}
		type = CMarketStream::kStreamTypeBond;
		return true;
	}
	
    bool CMarketStream::FilterDealList(xQBMarketStreamList_c* pLstdata, std::vector<ParsedStreamInfo>& batch)
	{
		// 初步过滤,静态函数
		uint32_t unDay = 0;
		int ncount = 0;
		if (pLstdata == nullptr || pLstdata->m_List.size() <= 0)
			return false;
		batch.clear();
		batch.reserve(pLstdata->m_List.size());
		CMarketStreamBase::kStreamType type = CMarketStreamBase::kStreamTypeBond;
		std::list<xQBMarketStreamUnit_c>::iterator it = pLstdata->m_List.begin();
		for (; it != pLstdata->m_List.end(); ++it)
		{
			const xQBMarketStreamUnit_c& msu = *it;
			if (CMarketStream::IsDealValid(msu, unDay, type))//NCD权限检查
			{
				BASSERTS(unDay > 0, "");
				ParsedStreamInfo si;
				si.unDay = unDay;
				si.type = type;
				detail::MSUtil::DecodeMarketStreamUnit(si.info, msu);
				batch.push_back(si);
				ncount++;
			}
		}
		return ncount > 0;
	}

    bool CMarketStream::FilterCFETSDealList(xQBMarketStreamList_c* pLstdata, std::vector<ParsedStreamInfo>& batch, uint32_t DealType)
	{
		// 初步过滤,静态函数
		int ncount = 0;
		if (pLstdata == nullptr || pLstdata->m_List.size() <= 0)
			return false;
		batch.clear();
		batch.reserve(pLstdata->m_List.size());
		CMarketStreamBase::kStreamType type = CMarketStreamBase::kStreamTypeCFets;
		char sCid[16] = { 0 };
		FMTBUF(sCid, "%d", DealType);
		std::list<xQBMarketStreamUnit_c>::iterator it = pLstdata->m_List.begin();
		for (; it != pLstdata->m_List.end(); ++it)
		{
			const xQBMarketStreamUnit_c& msu = *it;
			uint32_t unDay = qb::base::IntDate::timeToInt(qMax(msu.m_create_time, msu.m_modify_time));
			BASSERTS(unDay > 0, "company_id:%s id:%s",msu.m_company_id,msu.m_body2.m_id);
			ParsedStreamInfo si;
			si.unDay = unDay;
			si.type = type;
			detail::MSUtil::DecodeMarketStreamUnit(si.info, msu);
			FIELDCOPY(si.info.m_company_id, sCid);
			batch.push_back(si);
			ncount++;
		}
		return ncount > 0;
	}

	bool CMarketStream::UpdateMarketInfoMap(const MarketStreamInfo& record, int mktype, int stattype, CMarketInfoMap& infoMap)
	{
		// GetMarketInfoKey 用于获取计数统计的key
		// 如果检测失败,该函数返回-1,那么将不用于统计
		//typedef CMarketStream::MarktType MarktType;
		//typedef CMarketStream::kStatType kStatType;
		int nIndex = CBondContainer::instance().GetBondIndex(record.m_body.m_bondkey, record.m_body.m_listedmarket);
		if (!CBondContainer::instance().IsValidIndex(nIndex))
			return false;

		const CBondInfo& bond = CBondContainer::instance().ElementAtR(nIndex);
		if (IsBondPassType(bond, (MarktType)mktype))
		{
			kStatType stat = (kStatType)stattype;
			int key = GetMarketInfoKey(stattype, mktype, bond);
			if (key >= 0)
			{
				infoMap.m_map[key] += 1;
				return true;
			}
		}
		return false;
	}



CMarketStream::CMarketStream() :
m_tag(OBJECT_TAG),//TODO 干啥的？
m_nextid(1),
m_last(nullptr)
{

}

CMarketStream::~CMarketStream()
{
	SCOPED_MUTEX(m_lock);
	m_last = nullptr;
	for (DayDealMap::iterator it = m_deal_map.begin(); it != m_deal_map.end(); ++it)
	{
		RTDealList* dealList = it->second;
		if (dealList)delete dealList;
	}
	m_deal_map.clear();
}


CMarketStream& CMarketStream::cfectsStream()
{
	static CMarketStream s_cfets;
	return s_cfets;
}

CMarketStream& CMarketStream::exchangeStream()
{
	static CMarketStream s_ex;
	return s_ex;
}

CMarketStream& CMarketStream::bondStream()
{
	static CMarketStream s_bond;
	return s_bond;
}

bool CMarketStream::Clear(uint32_t uTime)
{
	SCOPED_MUTEX(m_lock);
	m_last = nullptr;
	if (uTime <= 0)
	{
		DayDealMap::iterator it = m_deal_map.begin();
		for (; it != m_deal_map.end();++it)
		{
			RTDealList* rtlist = it->second;
			if (rtlist)
				delete rtlist;
		}
		m_deal_map.clear();
	}
	else
	{
		DayDealMap::iterator it = m_deal_map.find(uTime);
		if (it != m_deal_map.end())
		{
			RTDealList* rtlist = it->second;
			if (rtlist)
				delete rtlist;
			m_deal_map.erase(it);
		}
	}

	return true;
}

int CMarketStream::GetVersion(int companyID)
{
	SCOPED_MUTEX(m_lock);
	RTDealList* dealList = GetOrCreateList(GetBondDealInvalidateDay());
	if (dealList)
		return dealList->GetVersion(companyID);
	return 0;
}

int CMarketStream::GetRecords(IRealTimeDealModel* keys, int& pos, int count, void* records)
{
	SCOPED_MUTEX(m_lock);
	RTDealList* dealList = GetOrCreateList(GetBondDealInvalidateDay());
	if (dealList)
		return dealList->GetRecords(keys, pos, count, records);
	return 0;
}
bool CMarketStream::GetMarketStreamStatByKey(const char * szCombBondKey, MarketStreamRec& stat, uint32_t timeDay)
{
	uint32_t UnDay = timeDay <= 0 ? GetBondDealInvalidateDay() : timeDay;
	SCOPED_MUTEX(m_lock);
	RTDealList* rdlist = GetList(UnDay);
	if (rdlist)
	{
		return rdlist->GetStat(szCombBondKey, stat);
	}
	return false;
}
bool CMarketStream::GetBondDeal(const char * szCombBondKey, const char* szCompanyId, MarketStreamInfo& record, time_t t, uint32_t day)
{
	/*
		用以替代CMarketStream中的GetMarketStreamByKey和GetNearByMarketStream接口
		时间复杂度:O(N)
		算法优化:map<szCombBondKey,list,DealIndex> 存起来;
	*/
	if (szCombBondKey == nullptr || szCompanyId == nullptr)
		return false;
	char szBondKey[32] = { 0 };
	char szListedMarket[32] = { 0 };
	{
		CBondContainer::ScopedLock l(CBondContainer::instance());
		int nBondIndex = CBondContainer::instance().GetBondIndex(szCombBondKey);
		if (nBondIndex < 0)
			return false;
		const CBondInfo& bi = CBondContainer::instance().ElementAtR(nBondIndex);
		FIELDCOPY(szBondKey, bi.GetBondKey());
		FIELDCOPY(szListedMarket, bi.GetListedMarket());
	}
	
	uint32_t UnDay = day <= 0 ? GetBondDealInvalidateDay() : day;

	SCOPED_MUTEX(m_lock);
	
	RTDealList* rdlist = GetList(UnDay);
	if (rdlist == nullptr)
		return false;
	
	return rdlist->FindRecord(szBondKey, szListedMarket, szCompanyId, record, t);
}

bool CMarketStream::GetBondDealBatch(const std::vector<BrokerKey>& brokerKeys, std::vector<MarketStreamInfo>& records, time_t t, uint32_t day)
{
	uint32_t UnDay = day <= 0 ? GetBondDealInvalidateDay() : day;
	records.reserve(brokerKeys.size());
	SCOPED_MUTEX(m_lock);
	RTDealList* rdlist = GetList(UnDay);
	if (rdlist)
	{
		for (int i = 0; i < (int)brokerKeys.size(); ++i)
		{
			MarketStreamInfo msi;
			const BrokerKey& key = brokerKeys[i];
			char szBondKey[32] = { 0 };
			char szListedMarket[32] = { 0 };
			const CBondInfo& bi = CBondContainer::instance().ElementAtR(key.m_nIndex);
			FIELDCOPY(szBondKey, bi.GetBondKey());
			FIELDCOPY(szListedMarket, bi.GetListedMarket());
			rdlist->FindRecord(szBondKey, szListedMarket, key.GetBrokerId().c_str(), msi, t);
			records.push_back(msi);
		}
		return true;
	}
	return false;

}
void CMarketStream::InsertCashData(xQBMarketStreamList_c* pLstdata, bool& ret)
{
	// 线程:后台线程ProcessQpidMessage
	// 作用:用来收取推动来的成交信息
	BASSERTS(IsObjectValid(), "");
	if (pLstdata && pLstdata->m_List.size() > 0)
	{
		// 初步过滤
        std::vector<ParsedStreamInfo> batch;
		if (FilterDealList(pLstdata, batch) && batch.size() > 0)
		{
			//log_warining("[STREAM][INFO]收到成交信息 %d 条,其中 %d 条无效", pLstdata->m_List.size(), (pLstdata->m_List.size() - batch.size()));
			SCOPED_MUTEX(m_lock);
			if (!IsObjectValid())
				return;
			int id = m_nextid++;
			for (int i = 0; i < (int)batch.size(); i++)
			{
				ParsedStreamInfo& si = batch[i];
				if (si.type==kStreamTypeBond)
					bondStream().GetOrCreateList(si.unDay)->InsertDirectly(si.info, si.unDay, id);
				else if (si.type == kStreamTypeExchange)
				{
					CMarketStream* pex = &exchangeStream();
					BASSERTS(pex, "");
					if (pex)
						pex->GetOrCreateList(si.unDay)->InsertDirectly(si.info, si.unDay, id);
				}
			}
		}
		ret = batch.size() > 0 ? true : false;
	}
}

void CMarketStream::InsertCFETSCashData(xQBMarketStreamList_c* pLstdata, uint32_t DealType, bool& ret)
{	// 线程:后台线程ProcessQpidMessage
	// 作用:用来收取推动来的成交信息
	if (pLstdata && pLstdata->m_List.size() > 0)
	{
		// 初步过滤
        std::vector<ParsedStreamInfo> batch;
		if (FilterCFETSDealList(pLstdata, batch, DealType) && batch.size() > 0)
		{
			log_debug("[STREAM][CFETS][INFO]收到成交信息 %d 条,其中 %d 条无效", pLstdata->m_List.size(), (pLstdata->m_List.size() - batch.size()));
			SCOPED_MUTEX(m_lock);
			if (!IsObjectValid())
				return;
			int id = m_nextid++;
			for (int i = 0; i < (int)batch.size(); i++)
			{
				ParsedStreamInfo& si = batch[i];
				if (si.type == kStreamTypeCFets)
					GetOrCreateList(si.unDay)->InsertDirectly(si.info, si.unDay, id);
			}
		}
		ret = batch.size() > 0 ? true : false;
	}
}

bool CMarketStream::IsObjectValid()const
{
	return m_tag == OBJECT_TAG;
}
CMarketStream::RTDealList* CMarketStream::GetOrCreateList(DayType day)
{
	SCOPED_MUTEX(m_lock);
	if (m_last && m_last->GetDay() == day)
		return m_last;

	
	DayDealMap::iterator it = m_deal_map.find(day);
	if (it != m_deal_map.end())
	{
		m_last = it->second;
	}
	else
	{
		DayStateMap::iterator sit = m_deal_states.find(day);
		int64 state = (sit == m_deal_states.end()) ? 0 : sit->second;
		state++;
		m_deal_states[day] = state;
		m_last = new RTDealList(day,state);
		m_deal_map[day] = m_last;
	}

	return m_last;
}
CMarketStream::RTDealList* CMarketStream::GetList(DayType day)
{
	if (day <= 0)day = GetBondDealInvalidateDay();
	DayDealMap::iterator it = m_deal_map.find(day);
	if (it != m_deal_map.end())
		return it->second;
	return nullptr;
}
namespace{
	bool IsCompanyIdPassed(const char* companyId, int corp)
	{
		int nTempCorp = 0;
		if (strcmp(companyId, "c") == 0)
			nTempCorp = 'c';
		else if (strcmp(companyId, "e") == 0)
			nTempCorp = 'e';
		else
			nTempCorp = atoi(companyId);

		if (corp == 0)
		{//成交数据 0为所有Broker
			if (!(nTempCorp > 0 && nTempCorp <= 6) || nTempCorp == 98) return false;
			else return true;
		}
		return corp == nTempCorp;
	}
}
bool CMarketStream::GetMarketStreamList(const MarkStreamListArg& arg, MSInfoList& mkList, MSRecMap* mkRecordMap)
{
	/*
		用以替换CMarketStream中的以参数形式返回CMarketStreamlst的接口
		经过调查,绝大部分不需要CMarketStreamlst中的m_Record部分,为避免性能损失,将此拆成两部分,第二部分可选

		过滤总共分为四个步骤:
		1.按照日期过滤,重复日期自动忽略跳过
		2.按照MarketStreamInfo.m_company_id过滤,
			如果arg.corp<0则直接跳过,如果arg.corp==0,检查m_company_id正确性
			其他情况,只有当arg.corp==atoi(MarketStreamInfo.m_company_id)才通过
		3.按照债券有效性和MarktType过滤(隐藏的)
			债券有效性是必须的,MarktType则不是必须的
			如果arg.mark_type<0则对MarktType过滤直接通过,表示用户没有设置过
		4.按照emMktStreamPage类型过滤
			这也是可选的,设置-1将直接忽略
	*/
	int ncount = 0;
    const std::vector<uint32_t>& days = arg.days;
    std::vector<DealIndex> indexes;
	SCOPED_MUTEX(m_lock);
	for (int i = 0; i < (int)days.size(); i++)
	{
		if (i > 0 && IsExistInVec(days, i, days[i]))//日期去重
			continue;
		indexes.clear();
		RTDealList* rdlist = GetList(arg.days[i]);
		if (rdlist == nullptr)
			continue;
		IRecordMap* values = rdlist->GetKeys(indexes);
		for (int k = 0; k < (int)indexes.size(); k++)
		{
			const void* ptr = values->FindRecord(indexes[k]);
			if (ptr == nullptr)
				continue;
			const Record& rd = *((const Record*)ptr);
			if (arg.corp >= 0 && !IsCompanyIdPassed(rd.m_company_id,arg.corp))
				continue;
			
			int nIndex = CBondContainer::instance().GetBondIndex(rd.m_body.m_bondkey, rd.m_body.m_listedmarket);
			if (!CBondContainer::instance().IsValidIndex(nIndex))
				continue;
			const CBondInfo& bond = CBondContainer::instance().ElementAtR(nIndex);
			if (arg.mark_type >= 0 && !IsBondPassType(bond, (MarktType)arg.mark_type))
				continue;
			if (arg.page_type>=0 && !IsBondPassPage(bond, (emMktStreamPage)arg.page_type))
				continue;
			ncount++;
			mkList.push_back(rd);
			if (mkRecordMap && arg.page_type != mktPage_NCD)
			{
				MarketStreamRec stat;
                std::string combBondKey = bond.GetCombBondKey();
				if (mkRecordMap->find(combBondKey) == mkRecordMap->end() &&
					rdlist->GetStat(combBondKey, stat))
				{
					// 统计信息按照债券key来覆盖插入
					mkRecordMap->insert(MSRecMap::value_type(combBondKey, stat));
				}
			}
		}
	}
	return ncount > 0;
}

bool CMarketStream::GetTopRecords(const MarketStreamTopArg& arg, CMarketToprec &MarketInfo)
{
	// 用以替换CMarketStream中返回CMarketToprec的接口
	// 用途:获取选定日期和指定类型的所有成交信息中,最活跃的前指定个数的成交信息
    const std::vector<uint32_t>& days = arg.days;
	CMarketToprec RecordInfo;
	SCOPED_MUTEX(m_lock);
	if (arg.mark_type < 0)
	{
		detail::RealTimeDealStat statmap;
		for (int i = 0; i < (int)days.size(); i++)
		{
			if (i > 0 && IsExistInVec(days, i, days[i]))
				continue;
			RTDealList* rdlist = GetList(arg.days[i]);
			if (rdlist)
				statmap.AddStatMap(rdlist->GetStatObject());
		}
		return detail::MSUtil::SortMarketRecord(statmap.GetStats(), RecordInfo, MarketInfo, arg.count);
	}
	else
	{
		std::map<std::string, MarketStreamRec> mapRecordTp;
		for (int i = 0; i < (int)days.size(); i++)
		{
			if (i > 0 && IsExistInVec(days, i, days[i]))
				continue;
			MSInfoList mkList;
			MSRecMap mkMap;
			MarkStreamListArg mkArg;
			mkArg.days.push_back(days[i]);
			mkArg.mark_type = arg.mark_type;
			GetMarketStreamList(mkArg, mkList, &mkMap);
			if (mkMap.size() > 0){
				MarketInfo.SetSize(mkList.size() + MarketInfo.m_nTotal);
				for (std::map<std::string, MarketStreamRec>::iterator itrRe = mkMap.begin(); itrRe != mkMap.end(); itrRe++){
					detail::MSUtil::InsertMarketRecord(itrRe->first.c_str(), itrRe->second, mapRecordTp);
				}
			}
		}
		if (mapRecordTp.size() < 0) return false;
		return detail::MSUtil::SortMarketRecord(mapRecordTp, RecordInfo, MarketInfo, arg.count);
	}
}
bool CMarketStream::GetMarketInfo(const MarketStreamInfoArg& arg, CMarketInfoMap& infoMap)
{
	// 用以替换CMarketStream中的返回CMarketInfoMap的接口
	// 用途:根据各种条件,对选定日期的成交信息进行计数统计
	BASSERTS(arg.stat_type > kStatTypeMin && arg.stat_type < kStatTypeMax, "");
	SCOPED_MUTEX(m_lock);
	int ncount = 0;
    const std::vector<uint32_t>& days = arg.days;
    std::vector<DealIndex> indexes;
	for (int i = 0; i < (int)days.size(); i++)
	{
		if (i > 0 && IsExistInVec(days, i, days[i]))
			continue;
		RTDealList* rdlist = GetList(arg.days[i]);
		if (rdlist==nullptr)
			continue;
		IRecordMap* values = rdlist->GetKeys(indexes);
		for (int k = 0; k < (int)indexes.size(); k++)
		{
			const void* ptr = values->FindRecord(indexes[k]);
			if (ptr == nullptr)
				continue;
			const Record& rd = *((const Record*)ptr);
			if (UpdateMarketInfoMap(rd, (MarktType)arg.mark_type, arg.stat_type, infoMap))
				ncount++;
		}
	}
	infoMap.SetSize(ncount);
	return ncount > 0;
}

//////////////////////////////////////////////////////////////////////////

const MarketStreamInfo& CMarketStream::iterator::operator*()const
{
	return m_enumerator->GetAt(m_index);
}
const MarketStreamInfo* CMarketStream::iterator::operator->()const
{
	return &(m_enumerator->GetAt(m_index));
}
CMarketStream::Enumerator::Enumerator(CMarketStream* object, uint32_t day,bool all) :
/*m_scoped(object->m_lock),*/ m_values(nullptr), m_stats(nullptr)
{
	RTDealList* dealList = object->GetList(day);
	if (dealList)
	{
		m_values = dealList->GetKeys(m_indexes, all);
		m_stats = dealList->GetStatObject();
	}
}

CMarketStream::iterator CMarketStream::Enumerator::begin()const
{
	return CMarketStream::iterator(0, this);
}
bool CMarketStream::Enumerator::findrec(const std::string& combBondKey, MarketStreamRec& rec)const
{
	if (m_stats && m_stats->GetStat(combBondKey, rec))
		return true;
	return false;
}

const MarketStreamInfo& CMarketStream::Enumerator::GetAt(int nIndex)const
{
	// 在调用此接口之前,必须通过iterator it!=end();
	BASSERTS(m_values, "m_values 必须不为0,m_indexes.size()=%d", m_indexes.size());
	BASSERTS(nIndex >= 0 && nIndex < (int)m_indexes.size(), "nIndex:%d m_indexes.size()=%d",nIndex,m_indexes.size());
	return *((const MarketStreamInfo*)m_values->FindRecord(m_indexes[nIndex]));
}
